package org.example.mq.producer;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import java.nio.charset.StandardCharsets;
import java.util.Properties;
import lombok.extern.slf4j.Slf4j;
import org.example.mq.MqConfig;
import org.springframework.beans.factory.annotation.Autowired;

@Slf4j
public class MQProducer {

    protected String   gid;
    protected Producer producer = null;
    @Autowired
    private   MqConfig mqConfig;


    public void init() {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.NAMESRV_ADDR, mqConfig.getAddress());
        properties.put(PropertyKeyConst.AccessKey, mqConfig.getAccessKey());
        properties.put(PropertyKeyConst.SecretKey, mqConfig.getSecretKey());
        properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis,
                mqConfig.getSendMsgTimeoutMillis());
        // properties.put(PropertyKeyConst.GROUP_ID, gid);

        if (log.isInfoEnabled()) {
            log.info("准备启动MQ生产者, 生产者ID：" + gid);
        }
        producer = ONSFactory.createProducer(properties);
        if (producer != null && producer.isClosed()) {
            producer.start();
            if (producer.isStarted() && log.isInfoEnabled()) {
                log.info("MQ生产者已启动, 生产者ID：" + gid);
            }
        } else {
            throw new ONSClientException("producer对象为空！");
        }
    }

    public Boolean send(String message, String topic, String tag) {
        return send(message, topic, tag, null);
    }

    public Boolean send(String message, String topic, String tag, Properties userProperties) {
        byte[] body = null;
        body = message.getBytes(StandardCharsets.UTF_8);
        if (tag == null || tag.length() <= 0) {
            tag = "*";
        }

        long startTime = 0L;
        Message msg = new Message(topic, tag, body);
        msg.getUserProperties().putAll(userProperties);
        if (log.isInfoEnabled()) {
            log.info("MQ 开始发送消息，pid: " + gid + "，topid: " + topic + "，tag: " + tag
                    + "， message:" + message);
            startTime = System.currentTimeMillis();
        }

        SendResult result = producer.send(msg);

        if (log.isInfoEnabled()) {
            long time = System.currentTimeMillis() - startTime;
            log.info("MQ 发送结果，MessageId:" + result.getMessageId() + ",耗时 :" + time);
        }

        return result.getMessageId() != null && result.getMessageId().length() > 0;
    }

    public void destroy() {
        if (producer != null && producer.isStarted()) {
            producer.shutdown();
        }
    }
}
